package com.haozhen.homework

/**
  * @author haozhen
  * @email haozh@ync1.com
  * @date 2021/1/3  16:28
  */
object Work5 {
  case class Point(label:String,point:Array[Double])


  private def getDistance(x:Array[Double],y:Array[Double]): Double ={
    math.sqrt(x.zip(y).map(v=>math.pow(v._1-v._2,2)).sum)
  }

  private def getIndex(p:Array[Double],centers:Array[Array[Double]]):Int={
    val dist = centers.map(point=>getDistance(point,p))
    dist.indexOf(dist.min)
  }

  def main(args: Array[String]): Unit = {
    import org.apache.spark.sql.SparkSession

    val K = 3
    val minDist = 0.001
    val spark = SparkSession.builder().master("local[*]").appName("work5").getOrCreate()

    spark.sparkContext.setLogLevel("WARN")


    var modleData = spark.read
      .option("header","true")
      .option("delimiter",",")
      .option("inferschema","true")
      .csv("data/Iris.csv")

    val points = modleData.rdd.map {
      case r => {
        Point(r.getString(5), Array(
          r.getDouble(1),
          r.getDouble(2),
          r.getDouble(3),
          r.getDouble(4)
        ))
      }
    }
    points.cache()

    val centerPoints = points.takeSample(withReplacement = false,K).map(_.point)
    var tempDist = 1.0

    while (tempDist>minDist){
        val indexRDD = points.map(p=>(getIndex(p.point,centerPoints),(p.point,1.0)))

        def arrayAdd(x:Array[Double],y: Array[Double]):Array[Double] = x.zip(y).map(v=>v._1+v._2)

        val catalogRDD = indexRDD.reduceByKey((x,y)=>(arrayAdd(x._1,y._1),x._2+y._2))

        val newCenterPoints  = catalogRDD.map{
          case (index,(point,count)) =>(index,point.map(_/count))
        }.collectAsMap()

        val dist = for(i<- 0 until K) yield {
          getDistance(centerPoints(i),newCenterPoints(i))
        }
        tempDist = dist.sum

      for((key,value)<- newCenterPoints){
        centerPoints(key) = value
      }
      println("distSum = "+ tempDist+" ")
    }

    println("final centers:")

    centerPoints.foreach(x=>println(x.toBuffer))

    spark.stop()
  }

}
